package Data::Conveyor::App::Dispatch;

# $Id: Dispatch.pm 13653 2007-10-22 09:11:20Z gr $

use strict;
use warnings;
use Time::HiRes 'usleep';
use Data::Conveyor::Lock::Dispatcher;


our $VERSION = '0.02';


use base 'Class::Scaffold::App::CommandLine';


__PACKAGE__
    ->mk_framework_object_accessors(
        ticket_provider   => 'ticket_provider',
        ticket_dispatcher => 'dispatcher',
    )
    ->mk_scalar_accessors(qw(
        done stage_class dispatcher_sleep lockpath lockhandler
    ))
    ->mk_integer_accessors(qw(ticket_count));


use constant GETOPT => qw/lockpath=s/;


sub app_init {
    my $self = shift;
    $self->SUPER::app_init(@_);

    $self->delegate->make_obj('ticket_payload');

    # If several dispatchers are running, we want to know which log message
    # came from which process. Can be done only now that
    # Class::Scaffold::App->app_init will have instantiated the log singleton.
    $self->log->set_pid;

    $self->ticket_count(0) unless defined $self->ticket_count;
    $self->dispatcher_sleep($self->delegate->dispatcher_sleep || 10);
    $self->lockpath($self->delegate->lockpath);
    $self->delegate->control->filename($self->delegate->control_filename);
}


sub check_lockfile {
    my $self = shift;

    return 1 if $self->delegate->ignore_locks;

    $self->lockhandler ||
    $self->lockhandler(
        Data::Conveyor::Lock::Dispatcher->new(
            lockpath => $self->lockpath));

    $self->lockhandler->lockstate;
}


sub app_code {
    my $self = shift;
    $self->log->info("starting");
    $self->SUPER::app_code(@_);

    # keep EINTR from looping over into the next sleep call;
    # this also should rollback the interrupted transaction,
    # which is exactly what we want. -ac
    local $SIG{INT} = sub { exit };

    my $success;

    while (!$self->done) {

        # this could stay here
        unless ($self->check_lockfile) {
            $self->done(1);
            last;
        }

        $self->ticket_count_inc;
        $self->done(1) if $self->ticket_count >=
            $self->delegate->max_tickets_per_dispatcher;

        unless ($self->delegate->control->read) {
            $self->log->info("control returned false, exiting.");
            $self->done(1);
            last;
        }

        my $ticket;

        # If there aren't any tickets waiting to be processed, don't exit,
        # just sleep. We don't want to keep starting and stopping dispatcher
        # processes just because there are no more tickets for a few seconds.


        unless (defined($ticket = $self->ticket_provider->get_next_ticket(
                    [ $self->delegate->control->allowed_stages_keys ], $success))) {

            $self->log->info("sleep %ss", $self->dispatcher_sleep);
            sleep($self->dispatcher_sleep);
            next;
        }

        # XXX: $ticket->stage should already be a ticket stage value object,
        # so we'd only need to do $ticket->stage->name.

        #my $stage = $self->delegate->make_obj('value_ticket_stage')->new(
        #    value => $ticket->stage)->name;

        # Try to open the ticket - this can still fail if another dispatcher
        # process has already opened the ticket.

        # try_open sets the stage to aktiv_* and commits - we don't want that
        # any more in nic.at
        # we should try to get the db locks instead.
        $self->log_line($ticket, '>');
        if ($self->open_ticket($ticket)) {
            $success = 1;
            # $self->log_line($ticket, $success);
        } else {
            $success = 0;
            # $self->log_line($ticket, $success);
            # cool it a little
            usleep(200_000);
            next;
        }

        # Now we have an opened ticket; process it.
        $self->process_ticket($ticket);
    }
    $self->log->info("exiting");
}


sub log_line {
    my ($self, $ticket, $success) = @_;
    $self->log->info("%s [%s] [% 3s] %s",
        $ticket->ticket_no, $success, $ticket->nice, $ticket->stage->name);
}



sub open_ticket {
    my ($self, $ticket) = @_;
    $ticket->try_open;
}


sub process_ticket {
    my ($self, $ticket) = @_;
    $self->dispatcher->dispatch($ticket);
}


1;


__END__

{% USE p = PodGenerated %}

=head1 NAME

{% p.package %} - stage-based conveyor-belt-like ticket handling system

=head1 SYNOPSIS

    {% p.package %}->new;

=head1 DESCRIPTION

None yet. This is an early release; fully functional, but undocumented. The
next release will have more documentation.

=head1 METHODS

=over 4

{% p.write_methods %}

=back

{% p.write_inheritance %}

{% PROCESS standard_pod %}

=cut

